package datastore.api.impl.hbase;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Scanner;
import org.apache.hadoop.hbase.io.RowResult;
import org.apache.hadoop.hbase.util.Bytes;

import datastore.api.DatastoreAPI;
import datastore.api.Entity;
import datastore.api.Query;
import datastore.api.Reference;
import datastore.api.impl.ScannerIterable;

public class HBaseDatastore implements DatastoreAPI {

    HBaseConfiguration config = new HBaseConfiguration();
    HBaseAdmin admin;
    String appName;
    EntityStore entityStore;
    //private BaseStore<Scanner, BaseQuery<?>> primaryIndex;
    private List<Store<?, ?>> indicis;

    public HBaseDatastore() {
    }

    // /////////////////////////////////////////////////////////////
    // DatabaseAPI methods
    // /////////////////////////////////////////////////////////////
    
    public void init(String appName) {
        this.appName = appName;
        try {
            this.admin = new HBaseAdmin(config);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            throw new RuntimeException(e);
        }

        /*String[] classNames = new String[modelClasses.length];
        for (int i = 0; i < modelClasses.length; i++) {
            classNames[i] = modelClasses[i].getName();
        }*/
        try {
            init();
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    /* (non-Javadoc)
     * @see datastore.api.impl.hbase.DatastoreImp#put(datastore.db.Model)
     */
    public List<?> put(Entity ... entities) {
        try {
            List<Object> keys = new ArrayList<Object>();
            for (Entity entity : entities) {
                //TODO move to api class completely
                entityStore.put(entity.getModel());
                for (Store<?, ?> index : indicis) {
                    index.put(entity.getModel());
                }
                keys.add(entity.getModel().getKey());
            }
            return keys;
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    /* (non-Javadoc)
     * @see datastore.api.impl.hbase.DatastoreImp#delete(datastore.db.Model)
     */
    public void delete(Entity ... entities) {
        try {
            for (Entity entity : entities) {
                //TODO move to api class completely
                entityStore.delete(entity.getModel());
                for (Store<?, ?> index : indicis) {
                    index.delete(entity.getModel());
                }
            }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    /* (non-Javadoc)
     * @see datastore.api.impl.hbase.DatastoreImp#getById(java.lang.Class, long)
     */
    public List<?> get(Reference... refs) {
        try {
            ArrayList<Object> list = new ArrayList<Object>();
            for (Reference ref : refs) {
                Object model = entityStore.get(ref.getKey());
                if (model != null) {
                    list.add(model);
                }
            }
            return (List<?>) list;
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    /* (non-Javadoc)
     * @see datastore.api.impl.hbase.DatastoreImp#fetch(java.lang.Class, int, int)
     */
    public List<?> fetch(Query query, int limit, int offset) {
        try {
            final ArrayList<Object> all = new ArrayList<Object>();
            
            for (Object model : iterable(query, limit, offset)) {
                all.add(model);
            }
            return all;
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    @Override
    public Iterable<?> iterable(Query query, int limit, int offset) {
        try {
            //TODO query for more than kind
            Scanner scanner = ((KindIdxStore)indicis.get(0)).get(query.getQuery());
            ScannerIterable<Object> iterable = new ScannerIterable<Object>(scanner, limit, offset, 
                new ScannerIterable.RowResultHandler(){
                    @Override
                    public Object get(RowResult rowResult) throws Exception {
                        String row = Bytes.toString(rowResult.getRow());
                        String entityRow = EntityStore.unescape(EntityStore.getLastPart(row));
                        Object model = entityStore.get(entityRow);
                        return model;
                }});
            return iterable;
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    /* (non-Javadoc)
     * @see datastore.api.impl.hbase.DatastoreImp#getAppName()
     */
    public String getAppName() {
        return appName;
    }

    // /////////////////////////////////////////////////////////////

    /**
     * Setup hbase tables if needed
     */
    private void init() throws Exception {
        this.entityStore = new EntityStore(this);
        this.indicis = new ArrayList<Store<?,?>>();

        entityStore.init();
        
        KindIdxStore kindIdx = new KindIdxStore(this);
        kindIdx.init();
        indicis.add(kindIdx);
        //TODO: add reverse kind, property and reverse property indices
    }
}
